package com.zp.datax.core;

import com.zp.datax.core.function.ConsumerWrapper;
import com.zp.datax.core.model.DataxJobLog;
import com.zp.datax.core.model.DataxJobTrigger;
import com.zp.datax.core.result.ReturnT;
import com.zp.datax.core.util.ThreadPoolFactory;
import com.zp.datax.core.util.TimeUtil;
import com.zp.datax.service.DataxJobLogService;
import com.zp.datax.service.DataxJobTriggerService;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.scheduling.support.SimpleTriggerContext;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * job线程
 *
 * @author bruce
 * @date 2019-07-15
 */
@Slf4j
public class DataxJobThread {

    //长时任务分隔点
    private final static long ONE_HOURS = 3600 * 4L;
    private volatile boolean runTaskStop = false;
    private volatile boolean slowerTaskStop = false;
    private TimingWheel timingWheel;
    private DataxJobTriggerService dataxJobTriggerService;
    private DataxJobLogService dataxJobLogService;
    private final ConsumerWrapper<DataxJobLog, Exception> logSupplierWrapper = (jobLog) -> dataxJobLogService.add(jobLog);//记录日志
    private BlockingQueue<SlowerTask> slowerTasks_queue;


    public DataxJobThread(DataxJobTriggerService dataxJobTriggerService, DataxJobLogService dataxJobLogService) {
        this.dataxJobTriggerService = dataxJobTriggerService;
        this.dataxJobLogService = dataxJobLogService;
        this.slowerTasks_queue = new DelayQueue<>();
        this.timingWheel = new TimingWheel();
    }

    public void push(Integer taskId) {
        updateNextDataxJobTrigger(setNextFireTime(dataxJobTriggerService.searchBySysId(taskId)));
    }

    public void remove(Integer taskId) {
        TimingWheel.Item item = timingWheel.removeTriggerTask(taskId);
        if (Objects.isNull(item)) {
            //查找慢任务对列中是否有任务开启，有则删除
            slowerTasks_queue.removeIf(t -> t.getTaskId() == taskId);
            log.warn("+++++++++++++++ 将 taskId :: {} 移除计划", taskId);
        }

        DataxJobTrigger jobTrigger = DataxJobTrigger
                .builder()
                .taskId(taskId)
                .runStatus(DataxJobStatus.STOP.getValue())
                .build();
        updateDataxJobTrigger(jobTrigger);
    }

    public void destroy() {
        if (!runTaskStop) runTaskStop = true;
        timingWheel.destroy();
    }

    private void startTaskThread() {

        ThreadPoolExecutor pool = ThreadPoolFactory.INSTANCE.getThreadPool();
        pool.execute(() -> {
            while (!runTaskStop) {
                try {

                    //移除并返回头部的元数，如果为空，则返回null
                    Integer taskId = timingWheel.getTriggerQueue().take();
                    //异步线程调用，避免多个任务同时触发执行时排对等待
                    pool.submit(() -> triggerTask(taskId));

                } catch (Exception e) {
                    if (!runTaskStop) {
                        log.error("+++++++++++++++ : {}", e.getMessage());

                    }
                }
            }
        });

        pool.execute(() -> {
            while (!slowerTaskStop) {
                try {

                    checkSlowerTaskOrTrigger();

                } catch (Exception e) {
                    if (!slowerTaskStop) {
                        log.error("+++++++++++++++ : {}", e.getMessage());

                    }
                }
            }
        });

    }

    @SneakyThrows
    private void checkSlowerTaskOrTrigger() {

        SlowerTask take = slowerTasks_queue.take();

        long newRemainingTime = take.getRemainingTime() - ONE_HOURS;
        if (newRemainingTime > ONE_HOURS) {
            slowerTasks_queue.add(new SlowerTask(take.getTaskId(), newRemainingTime, ONE_HOURS));
        } else timingWheel.pushTriggerTask(take.taskId, take.getRemainingTime() - ONE_HOURS);


    }

    private void triggerTask(final Integer taskId) {


        DataxJobTrigger dataxJobTrigger = dataxJobTriggerService.searchBySysId(taskId);
        MDC.put("log.filename", dataxJobTrigger.getName());

        String params = dataxJobTrigger.getParams();
        String path = dataxJobTrigger.getPath();

        //设置条件
        Condition condition = new EndIdCondition(new StartIdCondition(new EndTimeCondition(new StartTimeCondition(dataxJobTrigger))));
        //设置处理器
        ExecProcessor processor = new ThreadExecProcessor(new PageExecProcessor(path, params), condition);
        //设置过虑器
        Handler idHandler = new IdFilterHandler();
        ResultFilterHandler resultHandler = new ResultFilterHandler();
        resultHandler.setHandler(idHandler);

        DataxJobLog dataxJobLog = DataxJobLog.builder()
                .createTime(LocalDateTime.now(ZoneId.systemDefault()).toString())
                .description("自动触发")
                .taskId(taskId)
                .realParams(params)
                .status(0)
                .build();

        ReturnT returnT;
        try {
            //调用脚本
            returnT = processor.exec(resultHandler, logSupplierWrapper, dataxJobLog);
            dataxJobTrigger = updateDataxLogStatus(returnT, dataxJobTrigger);
        } catch (Exception e) {
            dataxJobLog.setModifyTime(LocalDateTime.now(ZoneId.systemDefault()).toString());
            dataxJobLog.setMsg(e.getMessage());
            dataxJobLog.setStatus(500);
            dataxJobLogService.add(dataxJobLog);
        }

        updateNextDataxJobTrigger(setNextFireTime(dataxJobTrigger));

        MDC.remove("log.filename");
    }

    @SneakyThrows
    public DataxJobTrigger updateDataxLogStatus(ReturnT<Map<String, Object>> returnT, DataxJobTrigger dataxJobTrigger) {

        Map<String, Object> data = returnT.getData();

        int code = returnT.getCode();
        dataxJobTrigger.setRtvStatus(code);

        //因为日志采用了异步生成，可能造成更新无法获取日志id，所以暂停10s
        //方式一，好处：线程不会一直hold，坏处：可能无法获取日志id更新日志
        /*if (data.get("sysId") == null) {
            TimeUnit.SECONDS.sleep(10);
        }*/

        //方式二，好处：有大机会获取日志id，坏处：但线程会一直hold，可以通过设置一个最大值退出
        AtomicInteger count = new AtomicInteger(0);
        while (data.containsKey("dataxJobLog")
                && Objects.isNull(((DataxJobLog) data.get("dataxJobLog")).getSysId())
                && count.getAndIncrement() <= 300) {

            TimeUnit.SECONDS.sleep(1);
        }

        DataxJobLog dataxJobLog = (DataxJobLog) data.get("dataxJobLog");

        //抽取记录总数
        Integer total = 0;
        Integer failCount = 0;

        //成功才更新记录，否则只更新状态码
        if (code == 200) {
            String realParams = dataxJobLog.getRealParams();
            if (data.containsKey("startTime")) {
                dataxJobTrigger.setStartTime(Long.valueOf(Objects.toString(data.get("startTime"))));
            }
            if (data.containsKey("endTime")) {
                dataxJobTrigger.setEndTime(Long.valueOf(Objects.toString(data.get("endTime"))));
            }
            if (data.containsKey("total")) {
                total = Integer.valueOf(Objects.toString(data.get("total")));
            }
            if (data.containsKey("failCount")) {
                failCount = Integer.valueOf(Objects.toString(data.get("failCount")));
            }
            if (data.containsKey("maxid")) {
                String readMaxid = Objects.toString(data.get("maxid"));
                if ("0".equals(readMaxid)) {
                    dataxJobTrigger.setMaxid(null);
                } else {
                    Long maxid = Long.valueOf(readMaxid);
                    Long conditionMaxid = dataxJobTrigger.getMaxid();

                    if (maxid.compareTo(conditionMaxid) <= 0 && dataxJobTrigger.getUnit().compareTo(7) == 0) {
                        dataxJobTrigger.setMaxid(maxid);
                    } else if (dataxJobTrigger.getUnit().compareTo(0) == 0) {
                        dataxJobTrigger.setMaxid(maxid);
                    }
                }
            } else if (Objects.nonNull(realParams)
                    && realParams.contains("startId")
                    && dataxJobTrigger.getRange().compareTo(0) == 1
                    && dataxJobTrigger.getUnit().compareTo(7) == 0) {

                Integer range = dataxJobTrigger.getRange();
                Long maxid = dataxJobTrigger.getMaxid();
                if (total.compareTo(0) == 0) {//如同本次同步记录总数为0，则条件maxid不更新
                    dataxJobTrigger.setMaxid(null);
                } else if (range.compareTo(total) == 1) {//如果同步范围>本次读取的记录总数，可能原因：1、源表数据在本次同步的范围内可能有断续id（曾经源表有删除过数据）；2、源表数据id可能还未增长到maxid；
                    dataxJobLog.setMsg(String.format(
                            "同步成功 :: 同步范围[%d]>本次读取的记录总数[%d]，可能原因：\n1、源表数据在本次同步的范围内可能有断续id（曾经源表有删除过数据）；\n2、源表数据id可能还未增长到maxid[%d]；\n建议人工核对本次同步，以免同步数据有漏；"
                            , range, total, maxid, realParams));
                } else if (range.compareTo(total) == 0) {//同步范围=本次读取的记录总数，本次同步成功；
                } else if (dataxJobTrigger.getRange().compareTo(total) == -1) {//同步范围<本次读取的记录数据，可能原因：1、源表数据不规范（id不是主键或id不唯一）
                    dataxJobLog.setMsg(String.format(
                            "同步成功 :: 同步范围[%d]<本次读取的记录总数[%d]，可能原因：\n1、源表数据不规范（id不是主键或id不唯一）；\n建议人工核对本次同步，以免同步数据有误；"
                            , range, total, realParams));
                }
            }
        } else {
            dataxJobTrigger.reset();
        }

        dataxJobLog.setStatus(code);
        dataxJobLog.setModifyTime(LocalDateTime.now(ZoneId.systemDefault()).toString());
        dataxJobLog.setTotal(total);
        dataxJobLog.setFailCount(failCount);
        dataxJobLogService.update(dataxJobLog);

        log.info("更新日志 :: {}", dataxJobLog.getSysId());

        return dataxJobTrigger;

    }

    public ReturnT<String> pageExec(Integer taskId, String params) {
        DataxJobTrigger dataxJobTrigger = dataxJobTriggerService.searchBySysId(taskId);

        MDC.put("log.filename", dataxJobTrigger.getName());

        Map<String, String> pageParams = parserPageParams(params);
        if (!pageParams.isEmpty()) {
            if (pageParams.containsKey("startTime")) {
                long timeMillis = TimeUtil.convert(pageParams.get("startTime"));
                dataxJobTrigger.setBizStarTime(timeMillis);
            }
            if (pageParams.containsKey("endTime")) {
                long timeMillis = TimeUtil.convert(pageParams.get("endTime"));
                dataxJobTrigger.setBizEndTime(timeMillis);
            }
            if (pageParams.containsKey("startId")) {
                dataxJobTrigger.setMinid(Long.valueOf(pageParams.get("startId")));
            }
            if (pageParams.containsKey("endId")) {
                dataxJobTrigger.setMaxid(Long.valueOf(pageParams.get("endId")));
            }
        }


        String pathname = dataxJobTrigger.getPath();
        //设置处理器
        PageExecProcessor processor = new PageExecProcessor(pathname, params);
        //设置过虑器
        Handler idHandler = new IdFilterHandler();
        ResultFilterHandler resultHandler = new ResultFilterHandler();
        resultHandler.setHandler(idHandler);

        DataxJobLog dataxJobLog = DataxJobLog.builder()
                .createTime(LocalDateTime.now(ZoneId.systemDefault()).toString())
                .description("手动触发")
                .taskId(taskId)
                .realParams(params)
                .status(0)
                .build();

        //调用脚本
        ReturnT<Map<String, Object>> returnT = processor.exec(resultHandler, logSupplierWrapper, dataxJobLog);

        updateDataxJobTrigger(updateDataxLogStatus(returnT, dataxJobTrigger));

        MDC.remove("log.filename");
        return new ReturnT<>(returnT.getCode(), returnT.getMsg());
    }

    public void init() {

        timingWheel.start();
        pushAll();
        startTaskThread();
    }

    private void pushAll() {

        List<DataxJobTrigger> dataxJobTriggers = dataxJobTriggerService.searchAll();

        final ConsumerWrapper<DataxJobTrigger, Exception> consumerWrapper = this::updateNextDataxJobTrigger;
        dataxJobTriggers.stream()
                .filter(t -> t.getRunStatus() == DataxJobStatus.NORMAL.getValue())
                .map(this::setNextFireTime)
                .forEach(consumerWrapper.andThen(t -> {
                }));

    }

    private Map<String, String> parserPageParams(String params) {
        if (Objects.isNull(params) || "".equals(params)) {
            return new HashMap<>();
        }

        return Arrays.stream(params.split("[,]"))
                .map(t -> t.split("[=]"))
                .collect(Collectors.toMap(t -> t[0].trim(), t -> t[1].trim()));
    }

    /**
     * 获取下次执行时间距离当前秒数
     *
     * @param dataxJobTrigger
     * @return
     */
    private long getNextExecutionTime(@NonNull DataxJobTrigger dataxJobTrigger) {

        CronTrigger cronTrigger = new CronTrigger(dataxJobTrigger.getCron());
        TriggerContext context = buildTriggerContext(dataxJobTrigger);
        Date nextExecutionTime = cronTrigger.nextExecutionTime(context);
        return nextExecutionTime.getTime();
    }

    private DataxJobTrigger setNextFireTime(@NonNull DataxJobTrigger dataxJobTrigger) {

        long nextFireTime = dataxJobTrigger.getNextFireTime();
        dataxJobTrigger.setPrevFireTime(nextFireTime);
        nextFireTime = getNextExecutionTime(dataxJobTrigger);
        dataxJobTrigger.setNextFireTime(nextFireTime);

        return dataxJobTrigger;
    }

    private TriggerContext buildTriggerContext(@NonNull DataxJobTrigger dataxJobTrigger) {

        ZoneId zoneId = ZoneId.systemDefault();
        LocalDateTime now = LocalDateTime.now(zoneId);
        LocalDateTime lastScheduledExecutionTime = TimeUtil.convertLocalDataTime(dataxJobTrigger.getNextFireTime());//最后调度时间
        LocalDateTime lastActualExecutionTime = TimeUtil.convertLocalDataTime(dataxJobTrigger.getStartTime());//最后实际执行时间
        LocalDateTime lastCompletionTime = TimeUtil.convertLocalDataTime(dataxJobTrigger.getEndTime());//最后完成时间


        ////////////////////////////////////////////////////////////////////////////////////
        //
        // 此处逻辑防止起动job任务时，导致计算最近调用时间拉长
        //
        ////////////////////////////////////////////////////////////////////////////////////


        log.info("currentDate :: {}, lastScheduledExecutionTime :: {}, lastActualExecutionTime :: {}, lastCompletionTime :: {}"
                , now
                , lastScheduledExecutionTime
                , lastActualExecutionTime
                , lastCompletionTime);

        //当前时间<最后调度时间，则：最后调度时间=当前时间
        if (now.isBefore(lastScheduledExecutionTime)) {
            lastScheduledExecutionTime = now;
        }

        //当前时间>最后完成时间，则：最后完成时间=当前时间
        if (now.isAfter(lastCompletionTime)) {
            lastCompletionTime = now;
        }


        return new SimpleTriggerContext(
                TimeUtil.convertDate(lastScheduledExecutionTime)
                , TimeUtil.convertDate(lastActualExecutionTime)
                , TimeUtil.convertDate(lastCompletionTime));
    }

    private void updateNextDataxJobTrigger(@NonNull DataxJobTrigger dataxJobTrigger) {

        long nextFireTime = dataxJobTrigger.getNextFireTime();


        Integer taskId = dataxJobTrigger.getTaskId();

        log.info("////////////////////////////////////////////////////////////////////////////////////");
        log.info("//");
        log.info("// taskId :: {},下次调用时间 :: {}", dataxJobTrigger.getTaskId(), TimeUtil.parserT(nextFireTime));
        log.info("//");
        log.info("////////////////////////////////////////////////////////////////////////////////////");

        DataxJobStatus status = DataxJobStatus.matched(dataxJobTrigger.getRunStatus().intValue());

        long nextSeconds = getNextSeconds(dataxJobTrigger.getNextFireTime());

        switch (status) {
            case STOP:
            case NONE:
                dataxJobTrigger.setRunStatus(DataxJobStatus.NORMAL.getValue());
            case NORMAL:
                if (nextSeconds > ONE_HOURS) {
                    SlowerTask slowerJobTrigger = new SlowerTask(
                            taskId
                            , nextSeconds
                            , ONE_HOURS);

                    updateDataxJobTrigger(dataxJobTrigger);
                    slowerTasks_queue.add(slowerJobTrigger);
                } else {
                    updateDataxJobTrigger(dataxJobTrigger);
                    timingWheel.pushTriggerTask(taskId, nextSeconds);
                }
        }

    }

    private long getNextSeconds(long nextFireTime) {
        return TimeUnit.SECONDS.convert(nextFireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    public void addDataxJobTrigger(DataxJobTrigger dataxJobTrigger) {
        dataxJobTriggerService.add(dataxJobTrigger);
    }

    public void updateDataxJobTrigger(DataxJobTrigger dataxJobTrigger) {
        dataxJobTriggerService.update(dataxJobTrigger);
    }

    public void removeDataxJobTrigger(Integer taskId) {
        dataxJobTriggerService.remove(taskId);
    }

    @Getter
    @RequiredArgsConstructor(access = AccessLevel.PACKAGE)
    final static class SlowerTask implements Delayed {

        final int taskId;//任务id
        final long remainingTime;//距离任务执行剩余时间，单位秒
        long executeTime;//执行时间

        SlowerTask(int taskId, long remainingTime, long expireTime) {
            this(taskId, remainingTime);
            this.executeTime = TimeUnit.NANOSECONDS.convert(expireTime, TimeUnit.SECONDS) + System.nanoTime();
        }

        @Override
        public final long getDelay(TimeUnit unit) {
            return unit.convert(executeTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        }

        @Override
        public final int compareTo(Delayed o) {
            SlowerTask task = (SlowerTask) o;

            return Long.valueOf(remainingTime).compareTo(Long.valueOf(task.getRemainingTime()));
        }

    }

}
